package com.yihaodian.search.broker;

import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import com.yihaodian.api.common.Response;
import com.yihaodian.api.search.SearchException;
import com.yihaodian.api.search.SearchRequest;
import com.yihaodian.api.search.SearchResponse;
import com.yihaodian.cluster.ClusterService;
import com.yihaodian.index.meta.shard.Shard;
import com.yihaodian.search.SearchListener;
import com.yihaodian.search.SearchPhaseBroker;
import com.yihaodian.search.broker.shard.ShardSearchClient;
import com.yihaodian.search.broker.shard.ShardSearchRequest;
import com.yihaodian.search.controller.QueryController;
import com.yihaodian.search.controller.ShardDoc;
import com.yihaodian.search.fetch.FetchRequest;
import com.yihaodian.search.fetch.FetchSearchResult;
import com.yihaodian.search.query.QuerySearchResult;

public class QueryThenFetchBroker extends SearchPhaseBroker {

	QueryController queryController = new QueryController();
	volatile ShardDoc[] sortedShards;

	public QueryThenFetchBroker(ClusterService clusterService,
			ShardSearchClient client) {
		super(clusterService, client);
	}

	/**
	 * TODO: Compare search() && doExecute() (Synchronous && callback)
	 */
	public SearchResponse search(SearchRequest request) throws SearchException {
		return null;
	};

	@Override
	protected void doExecute(SearchRequest request,
			SearchListener<Response> listener) {
		new AsyncManager(request, new ThreadedListener<Response>(listener))
				.start();
	}

	class AsyncManager extends BaseAsyncManager<QuerySearchResult> {
		private final Map<Shard, QuerySearchResult> queryResults = searchCache
				.obtainQueryResults();
		private final Map<Shard, FetchSearchResult> fetchResults = searchCache
				.obtainFetchResults();
		private Map<Shard, int[]> docIdForShard;

		public AsyncManager(SearchRequest request,
				SearchListener<Response> listener) {
			super(request, listener);
		}

		@Override
		protected QuerySearchResult sendExecuteFirstPhase(
				ShardSearchRequest shardRequest) {
			/**
			 * Send the first query for each shard. TODO: Need to be async.
			 */
//			return client.executeQueryPhrase(shardRequest);
			return null;
		}

		@Override
		protected void processFirstPhaseResult(QuerySearchResult result) {
			queryResults.put(result.getShard(), result);
		}

		@Override
		protected void moveToSecondPhase() throws SearchException {
			sortedShards = queryController.sortDocs(queryResults.values());
			// TODO: Get docIdFroShard from queryResults
			docIdForShard = queryController.docIdsPerShard(sortedShards);
			final AtomicInteger counter = new AtomicInteger(
					docIdForShard.size());

			/**
			 * TODO: Listener to handle the response: 1. Handle the failure
			 * response. 2. Limit the request dispatch times.
			 */
			for (final Map.Entry<Shard, int[]> entry : docIdForShard.entrySet()) {
				ThreadPools.getThreadPool("search").execute(new Runnable() {
					@Override
					public void run() {
						ShardSearchRequest shardRequest = new ShardSearchRequest(
								entry.getKey().getShardId().getShardId(), request);
						executeFetch(shardRequest, entry.getValue(), counter);
					}
				});

			}
		}

		private void executeFetch(ShardSearchRequest shardRequest,
				int[] docIds, AtomicInteger counter) {
			FetchRequest fetchRequest = new FetchRequest();
//			fetchRequest.request = shardRequest;
			fetchRequest.setDocIds(docIds);
			fetchRequest.setSize(docIds.length);

			//TODO
//			FetchSearchResult fetchResult = client
//					.executeFetchPhrase(fetchRequest);
			FetchSearchResult fetchResult = null;
			fetchResults.put(fetchResult.getShard(), fetchResult);
			if (counter.decrementAndGet() == 0) {
				finish();
			}
		}

		private void finish() {
			SearchResponse response = queryController.merge(sortedShards,
					queryResults, fetchResults);
			listener.onResponse(response);
		}
	}
}
